package com.ds.lens.flink.monitor.common;

import com.google.common.collect.Lists;
import com.ds.lens.data.common.constant.*;
import com.ds.lens.data.common.document.DataPoint;
import com.ds.lens.data.common.document.ForecastData;
import com.ds.lens.data.common.document.Notify;
import com.ds.lens.data.common.document.Triggered;
import com.ds.lens.data.common.document.dto.MonitorEvent;
import com.ds.lens.data.common.document.dto.MonitorKey;
import com.ds.lens.data.common.document.dynamic.DynamicSinkBO;
import com.ds.lens.data.common.util.MetricUtils;
import com.ds.lens.flink.monitor.common.bo.Metric;
import com.ds.lens.flink.monitor.common.bo.MetricBundle;
import com.ds.lens.flink.monitor.common.bo.MonitorDefinition;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.springframework.util.CollectionUtils;

import java.math.BigDecimal;
import java.util.*;

import static com.ds.lens.data.common.constant.Constants.METRIC_VALUE;
import static com.ds.lens.data.common.constant.Constants.TIME_STAMP;

@Slf4j
public class BuilderUtil {

    public static final String DEFAULT_GROUP_VALUE = "*";
    private static final String SPLITER = "-";
    public static final String DEV_RATIO = "devRatio";
    public static final String IS_ANOMALY = "isAnomaly";

    public static Map<String, Object> parseGroupValue(String groupValue) {
        if (groupValue == null) {
            return null;
        }
        groupValue = groupValue.trim();
        if (BuilderUtil.DEFAULT_GROUP_VALUE.equals(groupValue) || groupValue.isEmpty() || !groupValue.matches("^\\[.*\\]$")) {
            log.error("unexpected group by value {}", groupValue);
            return null;
        }
        Map<String, Object> map = new HashMap<>();
        groupValue = groupValue.substring(1, groupValue.length() - 1);
        String[] arr = groupValue.split(",");
        for (String s : arr) {
            String[] kvArr = s.split("=");
            if (kvArr.length != 2) {
                log.error("unexpected group by value {}", groupValue);
                return null;
            }
            map.put(kvArr[0], kvArr[1]);
        }
        if (map.isEmpty()) {
            return null;
        }
        return map;
    }

    public static Metric buildMetric(Long monitorId, Integer version, String metricName, AggregatorEnum aggrMethod,
                                     Map<String, Object> groupBys, List<DataPoint> values, Long startTime, Long endTime) {
        Metric metric = new Metric(new MonitorKey(monitorId, version), metricName, aggrMethod);
        metric.setValues(values);
        metric.setGroupBys(groupBys);
        metric.setStartTime(startTime);
        metric.setEndTime(endTime);
        return metric;
    }

    public static MonitorEvent buildDynamicMonitorEvent(MonitorDefinition definition, Metric metric, Double threshold,
                                                        CompareOperatorEnum operatorEnum, MonitorStateEnum state,
                                                        DynamicSinkBO dynamicSinkBO) {
        MonitorEvent monitorEvent = buildMonitorEvent(definition, metric, threshold, operatorEnum, state);
        monitorEvent.setDynamicSinkBO(dynamicSinkBO);
        return monitorEvent;
    }

    public static MonitorEvent buildMonitorEvent(MonitorDefinition definition, Metric metric, Double threshold,
                                                 CompareOperatorEnum operatorEnum, MonitorStateEnum state) {
        Long startTime = metric.getStartTime();
        if (state == MonitorStateEnum.NO_DATA) {
            startTime -= definition.getIntervalTime() * 60000; // 连续两个窗口无数据，才报NO_DATA, 所以重新校准startTime
        }
        Triggered triggerContent = Triggered.builder()
                .state(state)
                .startTime(startTime)
                .endTime(metric.getEndTime())
                .dataPoint(metric.getValues())
                .groupValue(generateGroupValue(metric.getGroupBys()))
                .build();
        Notify notify = Notify.builder()
                .channel(definition.getNotification().getChannel())
                .channelContent(definition.getNotification().getChannelContent())
                .content(customize(definition.getNotification().getContent(), definition, metric))
                .title(customize(definition.getNotification().getTitle(), definition, metric))
                .build();
        return MonitorEvent.builder()
                .uid(UUID.randomUUID().toString())
                .detectionMethod(definition.getDetectionMethod())
                .monitorName(definition.getMonitorName())
                .monitorKey(definition.getMonitorState().getKey())
                .monitorCreateTime(definition.getVersionTime())
                .expression(definition.getQueryDSL())
                .metricName(definition.getMetricName())
                .aggrEnum(definition.getMetricAggr())
                .operatorEnum(operatorEnum)
                .metricFilter(definition.getFilters())
                .groupBys(metric.getGroupBys())
                .tags(definition.getTags())
                .timeStamp(System.currentTimeMillis())
                .state(EventStateEnum.valueOf(state.name()))
                .triggered(triggerContent)
                .notify(notify)
                .threshold(threshold)
                .interval(definition.getIntervalTime().longValue())
                .queryDelayTime(definition.getRetriveDelayTime().longValue())
                .retriveDuration(definition.getRetiveDuration().longValue())
                .metricAggr(definition.getMetricAggr().getValue())
                .scope(definition.getScope())
                .integrationId(definition.getIntegrationId())
                .metricFilter(definition.getFilters())
                .metricBy(definition.getGroupBys())
                .appId(definition.getMonitorState().getApplicationId())
                .appName(definition.getMonitorState().getApplicationName())
                .levelThresholdMap(definition.getLevelThresholdMap())
                .queryStartTime(startTime)
                .queryEndTime(metric.getEndTime())
                .eventSourceEnum(EventSourceEnum.LENS_EVENT)
                .isNotify(true)
                .build();
    }

    private static String generateGroupValue(Map<String, Object> map) {
        if (CollectionUtils.isEmpty(map)) {
            return DEFAULT_GROUP_VALUE;
        }
        List<String> list = new ArrayList<>(map.keySet());
        Collections.sort(list);
        StringBuilder builder = new StringBuilder().append("[");
        boolean hasNext;
        Iterator<String> itr = list.iterator();
        do {
            String key = itr.next();
            builder.append(key).append("=").append(map.get(key));
            if (hasNext = itr.hasNext()) {
                builder.append(",");
            }
        } while (hasNext);
        builder.append("]");
        return builder.toString();
    }

    private static String customize(String text, MonitorDefinition definition, Metric metric) {
        StringBuilder builder = new StringBuilder().append("$$").append("$$");
        if (!CollectionUtils.isEmpty(metric.getGroupBys())) {
            for (String key : metric.getGroupBys().keySet()) {
                builder.replace(2, builder.length() - 2, key);
                text = text.replaceAll(builder.toString(), metric.getGroupBys().get(key).toString());
            }
        }
        if (!CollectionUtils.isEmpty(definition.getIncludes())) {
            for (String key : definition.getIncludes().keySet()) {
                builder.replace(2, builder.length() - 2, key);
                text = text.replaceAll(builder.toString(), definition.getIncludes().get(key).toString());
            }
        }
        return text;
    }

    public static MetricBundle buildMetricBundle(MonitorDefinition definition, Map<String, ForecastData> forecastDataMap,
                                                 List<Map<String, Object>> outList, Long startTime, Long endTime) {
        MetricBundle bundle = new MetricBundle(definition.getMonitorState());
        bundle.setStartTime(startTime);
        bundle.setEndTime(endTime);
        if (!CollectionUtils.isEmpty(outList)) {
            for (Map<String, Object> metricMap : outList) {
                Metric metric = new Metric(definition.getMonitorState().getKey(), definition.getMetricName(), definition.getMetricAggr());
                metric.setStartTime(startTime);
                metric.setEndTime(endTime);
                List<String> groupByList;
                if ((groupByList = definition.getGroupBys()) != null) {
                    for (String groupField : groupByList) {
                        Object value = metricMap.remove(groupField);
                        if (value != null) {
                            metric.addGroupByField(groupField, value);
                        }
                    }
                }
                String timeStr = metricMap.remove(TIME_STAMP).toString();
                Long timeStamp = Long.valueOf(new BigDecimal(timeStr).toPlainString());
                Object data = metricMap.remove(METRIC_VALUE);
                if (data == null) {
                    String key = MetricUtils.generateKey(definition.getMetricName(), definition.getMetricAggr());
                    data = metricMap.remove(key);
                }
                Double value = new BigDecimal(data.toString()).doubleValue();

                Double alertThresholdUpper = null;
                Double alertThresholdLower = null;
                Double forecast = null;
                if (forecastDataMap != null) {
                    ForecastData forecastData = forecastDataMap.get(generateKey(metric) + timeStamp);
                    if (forecastData != null) {
                        alertThresholdUpper = forecastData.getAlertThresholdUpper();
                        alertThresholdLower = forecastData.getAlertThresholdLower();
                        forecast = forecastData.getForecastValue();
                    }
                }

                metric.addMetricValue(new DataPoint(value, timeStamp, null, alertThresholdUpper, alertThresholdLower, forecast));
                bundle.addMetric(metric);
            }
        } else {
            Metric metric = new Metric(definition.getMonitorState().getKey(), definition.getMetricName(), definition.getMetricAggr());
            metric.setStartTime(startTime);
            metric.setEndTime(endTime);
            bundle.addMetric(metric);
        }

        return sortDataPoint(bundle);
    }

    private static MetricBundle sortDataPoint(MetricBundle bundle) {
        if (CollectionUtils.isEmpty(bundle.getMetrics().values())) {
            return bundle;
        }
        for (Metric metric : bundle.getMetrics().values()) {
            if (!CollectionUtils.isEmpty(metric.getValues())) {
                metric.getValues().sort(new DataPointComparator());
            }
        }
        return bundle;
    }

    public static String generateKey(Metric metric) {
        return generateKey(metric.getMonitorKey().getMonitorId(), metric.getMonitorKey().getVersion(), metric.getMetricName(), metric.getAggrMethod(), metric.getGroupBys());
    }

    public static String generateKey(Long monitorId, Integer version, String metricName, AggregatorEnum aggrMethod, Map<String, Object> groupMap) {
        StringBuilder builder = new StringBuilder().append(monitorId)
                .append(SPLITER).append(version).append(SPLITER).append(metricName);
        if (aggrMethod != null) {
            builder.append(SPLITER).append(aggrMethod.name());
        }
        if (!CollectionUtils.isEmpty(groupMap)) {
            builder.append(SPLITER);
            Iterator<String> itr = groupMap.keySet().iterator();
            boolean hasNext;
            do {
                String key = itr.next();
                Object value = groupMap.get(key);
                builder.append(value);
                if (hasNext = itr.hasNext()) {
                    builder.append(SPLITER);
                }
            } while (hasNext);
        }
        return builder.toString();
    }

    public static void main(String[] args) {
        List<DataPoint> values = Lists.newArrayListWithCapacity(5);
        Date date = new Date();
        for (int i = 0; i < 5; i++) {
            values.add(new DataPoint(1.5D, date.getTime(), null, 1D, 1D,1D));
            date = DateUtils.addMinutes(date, -1);
        }
        values.sort(new DataPointComparator());
        values.forEach(dataPoint -> System.out.println(DateFormatUtils.format(dataPoint.getTimeStamp(), "yyyy-MM-dd HH:mm:ss")));
    }
}
